package blue.mqtt4j.server.topic;

import blue.mqtt4j.core.handler.SenderListener;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author Jin Zheng
 * @since 1.0 2020-09-29
 */
public class TopicChannel
{
	private static Logger logger = LoggerFactory.getLogger(TopicChannel.class);

	private final ConcurrentMap<String, List<ChannelId>> map = new ConcurrentHashMap<>();
	private final ChannelGroup channelGroup;

	public TopicChannel()
	{
		this.channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
	}


	public void add(String topic, Channel channel)
	{
		List<ChannelId> idList = map.computeIfAbsent(topic, k -> new CopyOnWriteArrayList<>());
		idList.add(channel.id());

		channelGroup.add(channel);
	}

	public void write(String topic, MqttPublishMessage message)
	{
		List<ChannelId> idList = map.get(topic);
		if (idList == null || idList.isEmpty())
		{
			logger.info("{} channel is empty", topic);
			return;
		}
		for (ChannelId id : idList)
		{
			Channel channel = channelGroup.find(id);
			if (channel != null && channel.isActive())
			{
				channel.writeAndFlush(message).addListener(new SenderListener(message));
			}
		}
	}

}
